package com.apes.framework.plugin.mq;

import com.apes.framework.plugin.esb.api.SimpleRequest;
import com.apes.framework.plugin.event.Listener;
import com.apes.framework.plugin.mq.rocketmq.autoconfigure.RocketMQMessageListener;
import com.apes.framework.plugin.outbox.OutboxSubscriber;
import com.apes.framework.plugin.outbox.OutboxSubscriberManager;
import com.apes.framework.util.EventUtil;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 功能： OutboxEvent消费者
 *
 * @author xul
 * @create 2023-03-03 14:14
 */
@Service
public class OutboxEventConsumer {

    @Value("${apes.outbox.multichannel:false}")
    private boolean multichannel;

    private MessageChannel channel;

    private OutboxSubscriberManager subscriberManager;

    public OutboxEventConsumer(MessageChannel channel, OutboxSubscriberManager subscriberManager) {
        this.channel = channel;
        this.subscriberManager = subscriberManager;
    }

    public void initConsumer() {
        List<OutboxSubscriber> outboxSubscribers = subscriberManager.getAllOutboxSubscribers();
        addConsumer(outboxSubscribers);
    }

    private void addConsumer(List<OutboxSubscriber> subscribers) {
        if (!multichannel) return;
        subscribers.stream().forEach(subscriber -> {
            Consumer consumer = new Consumer() {
                private RocketMQMessageListener rocketMQMessageListener;

                @Override
                public String getTopic() {
                    return getSubscriberTopic(subscriber.getId());
                }

                @Override
                public boolean isBackup() {
                    return false;
                }

                @Override
                public int getPeriod() {
                    return 0;
                }

                @Override
                public boolean isTransaction() {
                    return false;
                }

                @Override
                public RocketMQMessageListener getRocketMQMessageListener() {
                    return rocketMQMessageListener;
                }

                @Override
                public void setRocketMQMessageListener(RocketMQMessageListener rocketMQMessageListener) {
                    this.rocketMQMessageListener = rocketMQMessageListener;
                }

                @Override
                public boolean consume(MQEvent ropEvent) {
                    return outboxMessageConsume(ropEvent.getMessage().getBody());
                }
            };
            channel.addConsumer(consumer);
        });
    }

    /**
     * 功能：消费
     **/
    @com.apes.framework.plugin.mq.annotation.Consumer(topic = "outbox.message", backup = false, transaction = false)
    public void consume(SimpleRequest request) {
        outboxMessageConsume(request.getData());
    }

    private boolean outboxMessageConsume(Object message) {
        EventUtil.sendSynEvent(this, "event:outbox.message.consume", message);
        return true;
    }

    /**
     * 功能：推送
     **/
    @Listener(topic = "event:outbox.message.publish")
    public void outboxMessagePublish(SimpleRequest request) {
        String topic = getSubscriberTopic(request.get("subscriber"));
        MQUtil.publish(topic, "发件箱消息", request.getData());
    }

    private String getSubscriberTopic(String subscriber) {
        String topic = "outbox.message";
        if (multichannel) {
            subscriber = subscriber.replaceAll("\\.", "_").replaceAll("\\:", "_");
            if (check(subscriber)) {
                topic += "_" + subscriber;
            }
        }
        return topic;
    }

    /**
     * 功能：命名规范检查
     **/
    public boolean check(String content) {
        return content.matches("^[%|a-zA-Z0-9_-]+$");
    }

}
